package com.zhiyuancode.mqack.worker.recorder;


import com.zhiyuancode.mqack.message.base.MqResendErrorCode;
import com.zhiyuancode.mqack.message.base.MqResendRuntimeException;
import com.zhiyuancode.mqack.message.base.addressAndMessage.AddressAndMessage4Object;
import com.zhiyuancode.mqack.message.base.message.BaseMessage;
import com.zhiyuancode.mqack.worker.RedisOperationsContext;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.RedisConnectionFailureException;

import java.io.IOException;
import java.util.*;

import static com.zhiyuancode.mqack.worker.Constants.*;


/**
 * 消息体有多个地址
 *
 * @author 杨志远
 * email faryangsh@163.com
 * Date    2018-05-02
 */
@Slf4j
public class Recorder4Object<T extends BaseMessage> extends BaseRecorder {
    Class<T> messageClass;
    protected final long resendTimeOutSecond;

    /**
     * @param redisOperationsContext redis操作操作对象
     * @param messageClass           消息类
     * @param resendTimeOutSecond    容忍消息ACK间隔时间(秒)
     */
    public Recorder4Object(RedisOperationsContext redisOperationsContext, Class<T> messageClass, long resendTimeOutSecond) {
        this.hashOperations = redisOperationsContext.getHashOperations();
        this.zSetOperations = redisOperationsContext.getZSetOperations();
        this.messageClass = messageClass;
        this.resendTimeOutSecond = resendTimeOutSecond;
    }

    public Recorder4Object(RedisOperationsContext redisOperationsContext, long resendTimeOutSecond) {
        this.hashOperations = redisOperationsContext.getHashOperations();
        this.zSetOperations = redisOperationsContext.getZSetOperations();
        this.resendTimeOutSecond = resendTimeOutSecond;
    }

    public Class<T> getMessageClass() {
        return messageClass;
    }

    /**
     * 保存已发的MQ记录
     *
     * @param
     */
    public void saveMessageRecord(T message) {
        try {
            // 记录该消息的msgId
            zSetOperations.add(RESEND_MSG_ID_COL, message.acquireMsgId(), System.currentTimeMillis());

            // 记录该消息的内容、地址、class类型
            Map<String, String> mqMsgMap = new HashMap<>(3);
            mqMsgMap.put(RESEND_MSG_ADDRESS, message.acquireMqAddress());
            mqMsgMap.put(RESEND_MSG_CONTENT, message.toJson());
            mqMsgMap.put(RESEND_MSG_CONTENT_CLASS, message.getClass().getName());
            hashOperations.putAll(RESEND_MSG_PREFIX + message.acquireMsgId(), mqMsgMap);
        } catch (RedisConnectionFailureException e) {
            log.error("failed save message record: msgId={}, e={}", message.acquireMsgId(), e.getMessage());
            throw new MqResendRuntimeException(MqResendErrorCode.REDIS_CONNECTION_EXCEPTION, e);
        } catch (Throwable e) {
            log.error("failed save message record: msgId={}, e={}", message.acquireMsgId(), e.getMessage());
            throw new MqResendRuntimeException(MqResendErrorCode.UNKNOW_EXCEPTION, e);
        }
    }

    /**
     * 查询待重发的消息
     *
     * @return
     */
    public List<AddressAndMessage4Object> queryResendMessage() throws IOException {
        long expireTime = System.currentTimeMillis() - resendTimeOutSecond * 1000;
        // 获取超过容忍时间外的消息记录
        Set<String> resendMsgId = zSetOperations.rangeByScore(RESEND_MSG_ID_COL, 0, expireTime);
        if (resendMsgId == null || resendMsgId.size() == 0) {
            return null;
        } else {
            List<AddressAndMessage4Object> mqList = new ArrayList<>();
            // 对所有超时的消息记录进行重发
            for (String msgId : resendMsgId) {
                Map<String, String> msgContent = hashOperations.entries(RESEND_MSG_PREFIX + msgId);
                String mqAddress = msgContent.get(RESEND_MSG_ADDRESS);
                String mqContent = msgContent.get(RESEND_MSG_CONTENT);
                // 不是该class类型的消息，不做重发
                if (!msgContent.get(RESEND_MSG_CONTENT_CLASS).equals(messageClass.getName())) {
                    continue;
                }
                AddressAndMessage4Object mqAddressAndMessage4Object;
                try {
                    mqAddressAndMessage4Object = AddressAndMessage4Object.builder()
                            .mqAddress(mqAddress)
                            .message(T.toObject(mqContent, messageClass))
                            .build();
                } catch (IOException e) {
                    throw e;
                }
                mqList.add(mqAddressAndMessage4Object);
            }
            return mqList;
        }
    }
}
